package com.zidongxiangxi.rabbit.producer.job;

import com.alibaba.fastjson.JSON;
import com.zidongxiangxi.rabbit.core.job.AbstractWorkThread;
import com.zidongxiangxi.rabbit.producer.config.ProducerResendConfig;
import com.zidongxiangxi.rabbit.producer.entity.Producer;
import com.zidongxiangxi.rabbit.producer.manager.ProducerManager;
import com.zidongxiangxi.rabbit.producer.manager.RabbitTemplateManager;
import com.zidongxiangxi.rabbit.producer.support.RabbitSequenceMessagePostProcessor;
import com.zidongxiangxi.rabbit.producer.util.ProducerUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.CollectionUtils;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.List;
import java.util.Objects;

/**
 * RabbitMq重发任务
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class ProducerResendThread extends AbstractWorkThread {
	private final Logger logger = LoggerFactory.getLogger(ProducerResendThread.class);

	private final String application;
	private final long workPeriod;
	private final int resendBatchSize;
	private final ProducerManager producerManager;
	private final RabbitTemplateManager rabbitTemplateManager;
	private final RabbitSequenceMessagePostProcessor rabbitSequenceMessagePostProcessor;

	public ProducerResendThread(
			ProducerResendConfig config,
			ProducerManager producerManager,
			RabbitTemplateManager rabbitTemplateManager,
			RabbitSequenceMessagePostProcessor rabbitSequenceMessagePostProcessor
	) {
		this.application = config.getApplication();
		this.workPeriod = config.getWorkPeriod();
		this.resendBatchSize = config.getBatchSize();
		this.producerManager = producerManager;
		this.rabbitTemplateManager = rabbitTemplateManager;
		this.rabbitSequenceMessagePostProcessor = rabbitSequenceMessagePostProcessor;
	}

	@Override
	public void work() {
		int size = resendMessage(resendBatchSize);
		if (size < resendBatchSize) {
			try {
				Thread.sleep(workPeriod);
			} catch (InterruptedException ignore) {}
		}
	}

	private int resendMessage(int batchSize) {
		List<Producer> list = producerManager.listSendingMq(application, 0, batchSize);
		if (CollectionUtils.isEmpty(list)) {
			return 0;
		}
		for (Producer producer : list) {
			try {
				producerManager.failSendMq(application, producer.getMessageId());
				RabbitTemplate rabbitTemplate = rabbitTemplateManager.getRabbitTemplate(producer.getHost(),
						producer.getPort(), producer.getVirtualHost());
				if (Objects.isNull(rabbitTemplate)) {
					logger.error("找不到RabbitTemplate，host:{}, port:{}, virtualHost:{}", producer.getHost(),
							producer.getPort(), producer.getVirtualHost());
					continue;
				}
				if (Objects.isNull(producer.getMessage())) {
					logger.error("消息的消息体为空：{}", JSON.toJSONString(producer));
					continue;
				}
				Message message = recoverMessage(producer.getMessage());
				CorrelationData correlationData = ProducerUtils.generateCorrelationData(producer.getApplication(),
						producer.getMessageId());
				rabbitTemplate.send(producer.getExchange(), producer.getRoutingKey(), message, correlationData);
			} catch (Throwable ignore) {}
		}
		return list.size();
	}

	private Message recoverMessage(byte[] messageBytes) throws IOException, ClassNotFoundException {
		ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(messageBytes));
		Message message = (Message)oin.readObject();
		return rabbitSequenceMessagePostProcessor.postProcessMessage(message);
	}
}
